package com.atguigu.chapter11;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.descriptors.Csv;
import org.apache.flink.table.descriptors.FileSystem;
import org.apache.flink.table.descriptors.Schema;

import static org.apache.flink.table.api.Expressions.$;

/**
 * Author: Pepsi
 * Date: 2023/8/24
 * Desc:
 */
public class Flink05_Table_source_File {
    public static void main(String[] args) {

        // 获取流执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 获取表执行环境，在这之前要获取流的执行环境作为参数传过来
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        // 元数据描述信息
        Schema schema = new Schema()
                .field("id", DataTypes.STRING())
                .field("ts", DataTypes.BIGINT())
                .field("vc", DataTypes.INT());

        // 创建动态表与文件管理
        tEnv
                // 用connect创建连接文件的，然后new一个fileSystem
                .connect(new FileSystem().path("input/sensor.txt"))
                // 因为这里需要传元数据信息，所以上面创建了对象
                .withSchema(schema)
                // 指定文件的格式
                .withFormat(new Csv())
                // 创建临时表的名字
                .createTemporaryTable("sensor");

        // 从刚才创建的表中读取数据，得到一个table对象
        Table table = tEnv.from("sensor");

        Table result = table
                .where($("id").isEqual("sensor_1"))
                .groupBy($("id"))
                .select($("id"), $("vc").sum().as("vc_sum"));

        Schema schema1 = new Schema()
                .field("id", DataTypes.STRING())
                .field("vc_sum", DataTypes.INT());
        tEnv
                .connect(new FileSystem().path("input/b.txt"))
                .withFormat(new Csv())
                .withSchema(schema1)
                .createTemporaryTable("abc");

        result.executeInsert("abc");
    }
}
